@drmingdrmer reviewed 2 files and all commit messages, and made 1 comment.
Reviewable status: 2 of 7 files reviewed, 1 unresolved discussion (waiting on dantengsky, SkyFan2002, TCeason, and zhyass).

src/meta/api/src/api_impl/table_api.rs line 379 at r1 (raw file):

                                if !existing_meta
                                    .engine
                                    .eq_ignore_ascii_case(&req.table_meta.engine)

why is this check done only when as_dropped is set? the codes does not tell the reason behind it.

Code quote:

                            if req.as_dropped {
                                // CTAS does not call construct_drop_table_txn_operations(),
                                // so validate its existing table here.
                                let existing_meta =
                                    self.get_pb(&TableId::new(*id.data)).await?.ok_or_else(|| {
                                        KVAppError::AppError(AppError::UnknownTableId(
                                            UnknownTableId::new(
                                                *id.data,
                                                "create or replace failed to find existing table meta",
                                            ),
                                        ))
                                    })?;
                                if !existing_meta
                                    .engine
                                    .eq_ignore_ascii_case(&req.table_meta.engine)

Request changes: the normal replacement path is close, but the invariant is not enforced at the publication point for two-phase CTAS. Temporary CTAS can still publish a cross-engine replacement.

1. Temporary CTAS has a check/commit race

TempTblMgr::create_table() checks the current engine, but always returns prev_table_id: None. Later, commit_table_meta() removes the orphan and unconditionally replaces whatever currently owns the visible name.

A long-running CTAS can therefore execute this sequence:

1. Prepare hidden FUSE table while the visible table is FUSE.
2. Another query in the same session drops/recreates the visible name as MEMORY.
3. CTAS commits and silently replaces MEMORY with FUSE.

Preserve the observed ID in the prepare result:

let existing_id = self.name_to_id.get(&desc).copied();

let new_table = match (existing_id, create_option) {
    // existing branches...
};

Ok(CreateTableReply {
    // ...
    prev_table_id: as_dropped.then_some(existing_id).flatten(),
    orphan_table_name,
})

Then make the commit validate before mutating either map:

let orphan_id = self
    .name_to_id
    .get(&orphan_desc)
    .copied()
    .ok_or_else(|| ErrorCode::UnknownTable(orphan_desc.clone()))?;

let current_id = self.name_to_id.get(&desc).copied();
if current_id != req.prev_table_id {
    return Err(ErrorCode::TableVersionMismatched(format!(
        "Temporary table {desc} changed while CTAS was running"
    )));
}

if let Some(current_id) = current_id {
    let current = self.id_to_table.get(&current_id).ok_or_else(|| {
        ErrorCode::Internal(format!("Missing temporary table metadata for {current_id}"))
    })?;
    let replacement = self.id_to_table.get(&orphan_id).ok_or_else(|| {
        ErrorCode::Internal(format!("Missing temporary table metadata for {orphan_id}"))
    })?;

    TableEngineMismatch::ensure(
        &req.name_ident.table_name,
        &current.meta.engine,
        &replacement.meta.engine,
    )
    .map_err(|e| ErrorCode::from(AppError::from(e)))?;
}

// Only mutate name_to_id and id_to_table after every validation succeeds.

The important part is to perform all checks before remove(&orphan_desc), so a rejected commit preserves both tables for diagnosis and cleanup.

2. Persistent CTAS validates too early

The engine check in create_table() happens while preparing the hidden table. The as_dropped branch does not bind the observed existing TableMeta sequence to that transaction, and commit_table_meta() does not compare the previous visible table engine with the hidden table engine.

Normal replacement is protected because construct_drop_table_txn_operations() adds a condition on the preloaded metadata sequence. CTAS must get equivalent protection in the transaction that publishes the hidden table.

At commit time, load the metadata for prev_table_id, compare it with the hidden table, and add its sequence to the same transaction:

let mut replacement_meta = tb_meta.ok_or_else(|| {
    KVAppError::AppError(AppError::UnknownTableId(
        UnknownTableId::new(table_id, "commit_table_meta"),
    ))
})?;

if let Some(prev_table_id) = req.prev_table_id {
    let prev_tbid = TableId::new(prev_table_id);
    let previous = self.get_pb(&prev_tbid).await?.ok_or_else(|| {
        KVAppError::AppError(AppError::UnknownTableId(
            UnknownTableId::new(prev_table_id, "commit_table_meta previous table"),
        ))
    })?;

    TableEngineMismatch::ensure(
        req.name_ident.table_name.as_str(),
        &previous.engine,
        &replacement_meta.engine,
    )
    .map_err(|e| KVAppError::AppError(e.into()))?;

    txn_req
        .condition
        .push(txn_cond_seq(&prev_tbid, Eq, previous.seq));
}

replacement_meta.drop_on = None;

This should run after checking that the history list still ends in prev_table_id and before constructing the publish transaction. The existing name, history-list, orphan, and replacement metadata conditions should remain.

Keeping the early check is useful for fast failure, but the commit-time check must be authoritative.

3. Centralize engine compatibility

The case-insensitive comparison and error construction are duplicated in the meta API and temporary-table manager. Put the policy next to TableEngineMismatch so all paths use identical semantics:

impl TableEngineMismatch {
    pub fn ensure(
        table_name: &str,
        existing_engine: &str,
        new_engine: &str,
    ) -> std::result::Result<(), Self> {
        if existing_engine.eq_ignore_ascii_case(new_engine) {
            return Ok(());
        }

        Err(Self::new(table_name, existing_engine, new_engine))
    }
}

This also gives tests one precise domain function to exercise.

4. Do not pass unkeyed preloaded metadata

construct_drop_table_txn_operations(table_id, preloaded_table_meta, ...) relies on a comment saying the metadata must belong to table_id. The type cannot enforce that. A future mismatched call could write metadata and clean up policy or MV references using the wrong table.

Prefer either letting the drop helper load its own metadata, or bind the key and value in one private type:

struct VersionedTable {
    id: TableId,
    meta: SeqV<TableMeta>,
}

async fn construct_drop_table_txn_operations(
    kv_api: &impl KVApi,
    existing: VersionedTable,
    req: &DropTableTxnReq,
    txn: &mut TxnRequest,
) -> Result<(u64, u64), KVAppError> {
    // existing.id and existing.meta cannot be mixed independently.
}

The current helper already has ten parameters. A small request object would also make invalid boolean combinations such as if_exists plus if_delete easier to avoid.

5. Add tests for the actual two-phase operation

The current meta test stops after create_table(as_dropped = true), the stream SQL test contains no AS SELECT, and the temporary-table test covers only immediate replacement. They do not exercise CTAS commit.

Please add:

• SQL: CREATE OR REPLACE TABLE stream_name AS SELECT ... fails and preserves the stream.
• SQL or manager test: temporary cross-engine CTAS fails and preserves the visible temporary table.
• Meta API test: prepare hidden CTAS, change the previous visible metadata, then verify commit returns TableEngineMismatch without changing the name mapping, history, or either table metadata.
• Exact error assertions such as assert_eq!(actual, expected) rather than only matching the enum variant.

The design rule should be: engine compatibility is checked and transactionally guarded where the visible name becomes associated with the replacement table.